package com.iot.fire.netty;

import com.iot.fire.service.*;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.net.InetSocketAddress;
import java.util.concurrent.*;

/**
 * @PackageName: com.iot.fire.tcp
 * @ClassName: TCPServer
 * @Description:
 * @Author: czl
 * @Date: 2025/11/9 17:13
 */
@Slf4j
@Data
@Component
@RequiredArgsConstructor
public class TCPServer {

    @Value("${netty.port:6883}")
    private int port;

    @Value("${netty.boss.threads:4}")
    private int bossThreads;

    @Value("${netty.worker.threads:32}")
    private int workerThreads;

    @Value("${netty.so.backlog:2048}")
    private int soBacklog;

    @Value("${netty.message.timeout:120}")
    private int messageTimeout;

    private final IDGatewayMessageService gatewayMessageService;
    private final IDWaterMessageService waterMessageService;
    private final IDWaterDeviceMessageService waterDeviceMessageService;
    private final ITDeviceService deviceService;
    private final RequestService requestService;

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;

    // 业务线程池，处理高并发消息
    private ExecutorService businessThreadPool;

    @PostConstruct
    public void start() throws InterruptedException {
        // 创建优化的线程组
        ThreadFactory bossThreadFactory = new DefaultThreadFactory("netty-boss", true);
        ThreadFactory workerThreadFactory = new DefaultThreadFactory("netty-worker", true);

        bossGroup = new NioEventLoopGroup(bossThreads, bossThreadFactory);
        workerGroup = new NioEventLoopGroup(workerThreads, workerThreadFactory);

        // 创建高并发业务线程池
        businessThreadPool = new ThreadPoolExecutor(
                64, // 核心线程数
                256, // 最大线程数
                120L, TimeUnit.SECONDS, // 空闲线程存活时间
                new LinkedBlockingQueue<>(50000), // 队列大小
                new DefaultThreadFactory("business-pool"),
                new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
        );

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();

                            // 添加空闲状态处理器
                            pipeline.addLast(new IdleStateHandler(
                                    messageTimeout, messageTimeout, messageTimeout, TimeUnit.SECONDS));

                            // 使用高性能的十六进制消息处理器
                            pipeline.addLast(new NettyServerHandler(businessThreadPool, gatewayMessageService,
                                    waterMessageService, waterDeviceMessageService,deviceService,requestService));
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, soBacklog)
                    .option(ChannelOption.SO_REUSEADDR, true)
                    .option(ChannelOption.SO_RCVBUF, 262144)  // 增大接收缓冲区
                    .option(ChannelOption.SO_SNDBUF, 262144)  // 增大发送缓冲区
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK,
                            new WriteBufferWaterMark(64 * 1024, 128 * 1024)); // 设置写缓冲水位

            ChannelFuture future = bootstrap.bind(new InetSocketAddress(port)).sync();

            log.info("✓ Netty服务器启动成功，端口：{} ，Boss线程数：{}，Worker线程数：{}", port, bossThreads, workerThreads);
            future.addListener(future1 -> {
                if (future1.isSuccess()) {
                    log.info("服务器绑定端口 {} 成功", port);
                } else {
                    log.error("服务器绑定端口失败: {}", future1.cause());
                }
            });
        } catch (InterruptedException e) {
            log.error("启动Netty服务器时发生异常: {}", e.getMessage());
            throw e;
        }
    }

    @PreDestroy
    public void destroy() {
        log.info("正在关闭Netty服务器...");

        // 优雅关闭业务线程池
        if (businessThreadPool != null) {
            businessThreadPool.shutdown();
            try {
                if (!businessThreadPool.awaitTermination(30, TimeUnit.SECONDS)) {
                    log.warn("业务线程池未在30秒内关闭，强制关闭");
                    businessThreadPool.shutdownNow();
                    if (!businessThreadPool.awaitTermination(10, TimeUnit.SECONDS)) {
                        log.error("业务线程池强制关闭失败");
                    }
                }
            } catch (InterruptedException e) {
                businessThreadPool.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }

        // 关闭Netty线程组
        if (bossGroup != null) {
            bossGroup.shutdownGracefully(5, 15, TimeUnit.SECONDS);
        }
        if (workerGroup != null) {
            workerGroup.shutdownGracefully(5, 15, TimeUnit.SECONDS);
        }

        log.info("Netty服务器已关闭");
    }
}
